package urlcounter;

import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

import urlcounter.URLCount.MyArrayWritable;

public class Stage1MapReduce {

	public static class TupleExtractorMapper extends
			Mapper<Object, Text, LongWritable, Text> {

		private final static LongWritable qId = new LongWritable();
		private final static Text url = new Text();

		public void map(Object key, Text inputLine, Context context)
				throws IOException, InterruptedException {
			String[] line = (inputLine.toString()).split(" ");
			qId.set(Integer.parseInt(line[0]));
			url.set(line[2]);
			context.write(qId, url);
		}
	}

	public static class DuplicationRemoverReducer extends
			Reducer<LongWritable, Text, LongWritable, MyArrayWritable> {
		private Set<Text> outSet = new HashSet<Text>();

		public void reduce(LongWritable key, Iterable<Text> urls, Context context)
				throws IOException, InterruptedException {
			MyArrayWritable arr = new MyArrayWritable(new String[] {});
			outSet.clear();
			Iterator<Text> it = urls.iterator();
			while (it.hasNext()) {
				outSet.add(new Text(it.next()));
			}
			Writable[] $ = new Text[outSet.size()];
			it = outSet.iterator();
			int i = 0;
			while (i < outSet.size()) {
				$[i++] = it.next();
			}
			arr.set($);
			context.write(key, arr);
		}
	}

}
